package com.atguigu.flink.datastreamapi.dirstribute;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/3
 */
public class Demo3_DisableOperatorChain
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //全局禁用 OperatorChain
        //env.disableOperatorChaining();

        SingleOutputStreamOperator<String> ds = env.socketTextStream("hadoop102", 8888).name("source");

        ds
            .map(x -> x).name("map1")
            .map(x -> x).name("map2")
            .map(x -> x).name("map3")
            //从当前算子开启一个新链，和之前的算子链断开
            //.startNewChain()
            // 当前算子禁用 算子链，既不和之前的算子合并，也不和之后的算子合并
            .disableChaining()
            .map(x -> x).name("map4")
            .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
